/*
 *  Copyright 2013 National Institute of Advanced Industrial Science and Technology
 *  
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *  
 *      http://www.apache.org/licenses/LICENSE-2.0
 *  
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 */

package org.sss.server;

import java.util.Iterator;
import java.util.UUID;

import org.slf4j.Logger;
import org.sss.mapreduce.Configuration;
import org.sss.mapreduce.Context;
import org.sss.mapreduce.GroupID;
import org.sss.mapreduce.IOConfiguration;
import org.sss.mapreduce.SideData;
import org.sss.mapreduce.SssException;
import org.sss.mapreduce.StorageNode;
import org.sss.mapreduce.Tuple;
import org.sss.mapreduce.TupleGroupScaner;
import org.sss.mapreduce.datatype.Packable;

public class ContextImpl implements Context {
  private static Logger logger = SssServer.getLogger(ContextImpl.class);

  private SssServerResource sssResource;
  private TaskSet taskSet;
  private Configuration config;
  private StorageNode node;
  private CounterProvider counter;

  public ContextImpl(SssServerResource resource,
                     StorageNode node,
                     TaskSet ts,
                     Configuration config,
                     CounterProvider counter) {
    this.sssResource  = resource;
    this.taskSet      = ts;
    this.config       = config;
    this.node         = node;
    this.counter      = counter;
  }
  public ContextImpl(SssServerResource resource,
      StorageNode node,
      TaskSet ts,
      Configuration config) {
    this.sssResource  = resource;
    this.taskSet      = ts;
    this.config       = config;
    this.node         = node;
  }

  @Override
  public Configuration getConfiguration() {
    return config;
  }

  @Override
  public <OK extends Packable, OV extends Packable>
  SideData<OK, OV> openSideData(GroupID gid, Class<OK> keyClass, Class<OV> valClass) {
    return openSideData(gid, keyClass, valClass, "default");
  }

  @Override
  public <OK extends Packable, OV extends Packable>
  SideData<OK, OV> openSideData(GroupID gid, Class<OK> keyClass, Class<OV> valClass, String sideDataType) {
    try {
      IOConfiguration ioConf = sssResource.getServerConfiguration().ioConf;
      if ("default".equals(sideDataType)) {
        sideDataType = config.get("sidedata.default.type", "ondemand");
      }
      if ("ondemand".equals(sideDataType)) {
        if (ioConf.stream_protocol) {
          logger.warn("Stream protocol does not support \"ondemand\" sidedata");
          return SideDataInAdvanceStream.create(sssResource, node, gid, taskSet.getEncodings(), keyClass, valClass);
        }
        else {
          return SideDataOnDemand.create(sssResource, node, gid, taskSet.getEncodings(), keyClass, valClass);
        }
      } else if ("inadvance".equals(sideDataType)) {
        if (ioConf.stream_protocol) {
          return SideDataInAdvanceStream.create(sssResource, node, gid, taskSet.getEncodings(), keyClass, valClass);
        }
        else {
          return SideDataInAdvance.create(sssResource, node, gid, taskSet.getEncodings(), keyClass, valClass);
        }
      } else {
        throw new RuntimeException("Unkown side data type: " + sideDataType);
      }
    } catch (Exception e) {
      throw new RuntimeException(e);
    }
  }

  @Override
  public <OK extends Packable, OV extends Packable>
  Iterable<Tuple<OK, OV>> getTupleReader(final GroupID gid, final Class<OK> keyClass, final Class<OV> valClass) {
    return new Iterable<Tuple<OK,OV>>() {
      @Override
      public Iterator<Tuple<OK, OV>> iterator() {
        try {
          return new TupleReader<OK, OV>(sssResource, node, gid, taskSet.getEncodings(), keyClass, valClass);
        } catch (SssException e) {
          throw new RuntimeException(e);
        }
      }
    };
  }

  @Override
  public void setTaskSetVariable(String key, Object value) {
    taskSet.set(key, value);
  }

  @Override
  public <T> T getTaskSetVariable(String key, Class<T> klass) {
    return taskSet.get(key, klass);
  }

  @Override
  public <OK extends Packable, OV extends Packable>
  void scanSideData(GroupID gid, Class<OK> keyClass, Class<OV> valClass, TupleGroupScaner<OK, OV> scaner) throws SssException {
    UUID jobid = UUID.randomUUID();
    ReadTask<OK, OV> task = new ReadTask<OK, OV>(taskSet, jobid, sssResource, config, node, gid, keyClass, valClass, scaner);
    task.execute();
  }

  @Override
  public void incrCounter(Enum<?> counterName, int val) {
    if (counter != null) {
      counter.updateCounter(counterName.name(), val);
    }
  }
  
}
